package com.zhenjun.dimension;

import com.zhenjun.domin.Order;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * @author wangzj
 * @description
 * @date 2020/8/17 23:04
 */
public class StreamingMain {
    public static void main(String[] args) throws Exception {
        //创建流式计算环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置时间属性为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //监控本地的9000端口
        //开启方式：nc -L -p 9000（先开端口然后启动程序）
        DataStream<String> text = env.socketTextStream("127.0.0.1", 9000, "\n");
        //1、将接收的数据进行查询到城市的名称(实时维表查询)
//        SingleOutputStreamOperator<Order> orderInfo = text.map(new DimSync());

        //2、预加载全量数据
        //TODO 代码运行报错，待解决问题
//        SingleOutputStreamOperator<Order> orderInfo = text.map(new WholeLoad());

        //3、LRU是一种缓存算法
        /**
         * unorderedWait参数说明
         * 1 输入数据，类型为DataStream
         * 2 继承RichAsyncFunction的类
         * 3 异步操作的超时时间
         * 4 超时的时间单位（毫秒）
         * 5 可以触发的异步I/O操作的最大数量
         */
        SingleOutputStreamOperator orderInfo = AsyncDataStream.unorderedWait(text, new LRU(), 1000, TimeUnit.MILLISECONDS, 100);

        //打印结果
        orderInfo.printToErr();

        env.execute("Stream Test");
    }
}
